/**
 * @author zhounan 
 * @date 2024年6月25日
 * @version V1.0
 */
package org.zn.quality.service.impl;

import java.sql.SQLException;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.stream.Collectors;

import javax.sql.DataSource;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.zn.datax.common.entity.ColumnInfo;
import org.zn.datax.common.entity.TableInfo;
import org.zn.etl.config.Constant;
import org.zn.etl.datasource.DataSourceFactoryHolder;
import org.zn.etl.jdbc.QueryHelper;
import org.zn.etl.job.entity.DataMapping;
import org.zn.etl.job.entity.EtlJob;
import org.zn.etl.job.entity.JobGroup;
import org.zn.etl.misc.DataSourceInfo2ConfigMap;
import org.zn.etl.misc.ExpressionHelper;
import org.zn.etl.misc.UuidUtils;
import org.zn.etl.service.EtlPersistService;
import org.zn.quality.dao.QualityResultMapper;
import org.zn.quality.entity.FaultsData;
import org.zn.quality.entity.QualityExecParam;
import org.zn.quality.entity.QualityStrategyEntity;
import org.zn.quality.entity.po.QualityResultPo;
import org.zn.quality.entity.po.QualitySampleRulePo;
import org.zn.quality.entity.po.QualityStatisticsRulePo;
import org.zn.quality.entity.po.QualityStrategyInfoPo;
import org.zn.quality.handler.AllRowStrategyImpl;
import org.zn.quality.handler.CompareDataUtils;
import org.zn.quality.handler.QualityHandler;
import org.zn.quality.service.QualityExecuteService;
import org.zn.quality.service.QualityManagerService;

import lombok.extern.slf4j.Slf4j;

/**
 * @author zhounan 
 * @date 2024年6月25日
 * @version V1.0
 */
@Slf4j
@Service
public class QualityExecuteServiceImpl implements QualityExecuteService{
	
	@Autowired
	private QualityManagerService qualityManagerService;
	
	@Autowired
	private QualityResultMapper qualityResultMapper;
	
	@Autowired
	private EtlPersistService  persistService;
	@Autowired
	private DataSourceFactoryHolder factoryHolder; 
	@Autowired
	private EtlPersistService	etlPersistService;
	
	@Autowired
	private ExpressionHelper expressionHelper; 
	
	@Autowired
	private  Map<String,QualityHandler>  qualityHandlerMap;
	
	@Autowired
	private AllRowStrategyImpl allRowStrategy;
	

	@Override
	public List<QualityResultPo> queryJobStrategyResult(Long jobid, Date startTime, Integer warning) {
		// TODO Auto-generated method stub
		if(startTime==null) {
			 Calendar calendar = Calendar.getInstance();
		        calendar.set(Calendar.MINUTE, 0);
		        calendar.set(Calendar.SECOND, 0);
		        calendar.set(Calendar.MILLISECOND, 0);
		        startTime =calendar.getTime();
		}
		
		return	qualityResultMapper.selectJobStrategyResult(jobid, startTime, warning);

	}
	
	
	public List<QualityResultPo> queryStrategyResult(Long strategyId,Date startTime,Integer warning){
		
		if(startTime==null) {
			 Calendar calendar = Calendar.getInstance();
		        calendar.set(Calendar.MINUTE, 0);
		        calendar.set(Calendar.SECOND, 0);
		        calendar.set(Calendar.MILLISECOND, 0);
		        startTime =calendar.getTime();
		}
		
		return qualityResultMapper.selectStrategyResult(strategyId, startTime, warning);

	}
	
	
	
	
	@Override
	public void executeAvailableStrategy(QualityExecParam param)  {
		
		List<QualityStrategyInfoPo>  list=  qualityManagerService.selectAvailableByType(param.getRuleType());
		
		for (QualityStrategyInfoPo ent : list) {
			
			try {
				
				log.info("start execute Strategy: {} ,type:",ent.getName(),ent.getRuleType());
				
				executeStrategy(ent.getId(), param.getParamMap());
				
				log.info("execute finish Strategy: {} ",ent.getName());

			} catch (Exception e) {
				
				log.error("execute error: {} ",ent.getName(),e);
			}
			
		}
		
	}
	
	

	@Override
	public Long executeStrategy(Long strategyId, Map<String, Object> paramMap) throws Exception {
		// TODO Auto-generated method stub

		QualityStrategyEntity qs=qualityManagerService.queryStrategyEntityById(strategyId);
		
		EtlJob job = persistService.queryEtlJobById(qs.getJobId());
		Map<String, String> sourceConfig=DataSourceInfo2ConfigMap.toMap(job.getSource());		
		Map<String, String> targetConfig=DataSourceInfo2ConfigMap.toMap(job.getTarget());
		
		DataSource sSource=factoryHolder.getFactory(job.getSource().getFactoryType()).getDataSource(job.getSource().getKeyName(), sourceConfig);
		DataSource tSource=factoryHolder.getFactory(job.getTarget().getFactoryType()).getDataSource(job.getTarget().getKeyName(), targetConfig);
		
		QualityResultPo qr=null;
		
		qs.setParamMap(paramMap);
		
		if("tableCount".equals(qs.getRuleType())) {
			qr=tableCount(sSource, tSource, job,qs);
		}
		else if("statistics".equals(qs.getRuleType())) {
			qr=statistics(sSource, tSource, job,qs);
		}
		else if("sample".equals(qs.getRuleType())) {
			qr=sampleRate(sSource, tSource, job,qs);
		}
		else if("allRows".equals(qs.getRuleType())) {
			qr=allRowStrategy.doStrategy(sSource, tSource, job,qs);
		}
		
		qr.setStrategyId(strategyId);
		//qr.setId(UuidUtils.generateIdString());
		
		qualityResultMapper.insert(qr);
	
		return qr.getId();
		
		
	}

	
	@Async
	public void executeJobQuality(String jobName,String ruleType,Map<String, Object> paramMap) throws Exception {
		
		EtlJob job = persistService.queryEtlJob(jobName);
		
		Map<String, String> sourceConfig=DataSourceInfo2ConfigMap.toMap(job.getSource());		
		Map<String, String> targetConfig=DataSourceInfo2ConfigMap.toMap(job.getTarget());
		
		DataSource sSource=factoryHolder.getFactory(job.getSource().getFactoryType()).getDataSource(job.getSource().getKeyName(), sourceConfig);
		DataSource tSource=factoryHolder.getFactory(job.getTarget().getFactoryType()).getDataSource(job.getTarget().getKeyName(), targetConfig);
		
		List<QualityStrategyEntity>   strategyList=qualityManagerService.queryJobStrategy(jobName);
		
		if(StringUtils.isNotEmpty(ruleType)) {
			strategyList=strategyList.stream().filter(it->  Objects.equals(ruleType, it.getRuleType())  ).collect(Collectors.toList());
		}

		for (QualityStrategyEntity stra : strategyList) {
		
			QualityResultPo qr=null;
			try {			
			stra.setParamMap(paramMap);
			
			if("tableCount".equals(stra.getRuleType())) {
				qr=tableCount(sSource, tSource, job,stra);
			}
			else if("statistics".equals(stra.getRuleType())) {
				qr=statistics(sSource, tSource, job,stra);
			}
			else if("sample".equals(stra.getRuleType())) {
				qr=sampleRate(sSource, tSource, job,stra);
			}
			else if("allRows".equals(stra.getRuleType())) {
				qr=allRowStrategy.doStrategy(sSource, tSource, job,stra);
			}
			
			//qr.setId(UuidUtils.generateIdString());
			qr.setStatus("finish");
			qualityResultMapper.insert(qr);

			} catch (Exception e) {
				// TODO: handle exception
				log.error("execute job:{} , strategy:{} ",jobName,stra.getName(),e);
				
			}
			
			

		}
		
	}
	
	@Async
	public void executeJobGroupQuality(String groupName,String ruleType, Map<String, Object> paramMap) throws Exception {
		
		JobGroup jg=etlPersistService.queryJobGroupBy(groupName);
		
		List<EtlJob> jobList=jg.getJobList();
		
		for (EtlJob etlJob : jobList) {
			
			executeJobQuality(etlJob.getJobName(),ruleType,paramMap);
			
		}
		
		
	}
	
	
	
	private QualityResultPo tableCount(DataSource sSource,DataSource tSource,EtlJob job,QualityStrategyEntity stra) throws SQLException {
		
		DataMapping dm=job.getDataMappings().get(0);
		
		TableInfo targetTable =dm.getTargetTable();
		TableInfo sourceTable =dm.getSourceTable();
		QualityResultPo qr=new QualityResultPo();
		qr.setStartTime(new Date());
		
		String tsql= new StringBuilder( "select count(1) total from ").append(targetTable.getSchema()).append(".").append(targetTable.getTable()).toString();
		
	
		Long tcount=  Long.valueOf(Optional.ofNullable( QueryHelper.queryMapList(tSource, tsql).get(0).get("total")).orElse("0").toString()) ;
		
		String ssql= new StringBuilder( "select count(1) total from ").append(sourceTable.getSchema()).append(".").append(sourceTable.getTable()).toString();
		
		Long scount=  Long.valueOf(Optional.ofNullable( QueryHelper.queryMapList(sSource, ssql).get(0).get("total")).orElse("0").toString()) ;

		qr.setFinishTime(new Date());

		Map<String, Object> resMap=new HashMap<>();
		resMap.put(targetTable.getSchema()+"."+targetTable.getTable(), tcount);
		resMap.put(sourceTable.getSchema()+"."+sourceTable.getTable(), scount);
		
		qr.setInspectResult(resMap);
		qr.setWarnning( Objects.equals(tcount, scount)?0:1);
		qr.setStrategyId(stra.getId());
		return qr;
		
		
	}
	
	
	private QualityResultPo sampleRate(DataSource sSource,DataSource tSource,EtlJob job,QualityStrategyEntity stra) throws SQLException {
		
		QualitySampleRulePo  sr =	stra.getSampleRule();
		
		Map<String, Object> context=stra.getParamMap();
		
		String side=sr.getSide();
		
		TableInfo table=null;
		
		if(Constant.TARGET.equals(side)) {
			table= job.getDataMappings().get(0).getTargetTable();
		}else {
			table =job.getDataMappings().get(0).getSourceTable();
		}
		
		context.put("table", table.getSchema()+"."+table.getTable());
		context.put("rate", sr.getRate());
		
		Long mount=	sr.getMonut();
		Integer rate=sr.getRate();
		Integer cardinality=sr.getRateCardinality();
		Integer round=sr.getRound();
		
		String sql=sr.getSql();

		DataMapping dm=job.getDataMappings().get(0);
		
		TableInfo targetTable =dm.getTargetTable();
		TableInfo sourceTable =dm.getSourceTable();
		
		List<Pair<ColumnInfo, ColumnInfo>>  cmps=	dm.getColumnMappingPairs();
		
		QualityResultPo qr=new QualityResultPo();
		qr.setStartTime(new Date());
		
		TableInfo main=sourceTable;
		TableInfo minor=targetTable;
		
		DataSource  mainSource=sSource;
		DataSource  minorSource=tSource;
		
		Map<String,String >  columnMapping=null;
		
		if(Constant.TARGET.equals(side)) {			
			main=targetTable;
			minor=sourceTable;
			
			mainSource=tSource;
			minorSource=sSource;
			
			columnMapping=cmps.stream().collect(Collectors.toMap( it-> it.getRight().getName(),v->v.getLeft().getName()));
		}else {
			columnMapping=cmps.stream().collect(Collectors.toMap( it-> it.getLeft().getName(),v->v.getRight().getName()));
		}
		
		String countSql= new StringBuilder( "select count(1) total from ").append(main.getSchema()).append(".").append(main.getTable()).toString();
		
		//Long  total=(Long) QueryHelper.queryMapList(sSource, countSql).get(0).get("total");
		
		Float sinlgeRate=1f;
		
		if(rate!=null&& rate>0) {
			
			if(cardinality==null)cardinality=1;
			if(round==null) round=1; 	

			sinlgeRate=sinlgeRate*rate/round;
			
			DecimalFormat fmt=new DecimalFormat("0.000000");
			String 	smapleRate=fmt.format(sinlgeRate);
			smapleRate=smapleRate.replaceAll("0+$", "");
			
			if(smapleRate.endsWith(".")) {
				smapleRate=smapleRate.substring(0, smapleRate.length()-1);
			}
			
			context.put("rate", smapleRate);
		}
		
		String warrpedSql= expressionHelper.parseExpression(sql).getValue(ExpressionHelper.map2EvalContext(context), String.class);

		final String mpk=main.getPk().toLowerCase();
		final String ipk=minor.getPk().toLowerCase();
		
		List<FaultsData> faultList=new ArrayList<>();

		for (int i = 0; i < round; i++) {
			
			List<Map<String, Object >>  data=QueryHelper.queryMapList(mainSource, warrpedSql);
			
			Map<Object, Map<String, Object >> dataMap= data.stream().collect(Collectors.toMap(it-> it.get(mpk), v->v));
			
			List<Object> mpkvList=new ArrayList<>(dataMap.keySet());
			
			int trun=1;
			
			trun=mpkvList.size()/100 + mpkvList.size()%100>0?1:0;

			StringBuilder minorSlq=new StringBuilder("select * from ").append(minor.getSchema()).append(".").append(minor.getTable()).append(" where ").append(ipk).append(" in (");
			
			for (int j = 0; j < trun; j++) {
				
				List<Object> slist= mpkvList.subList(j*100,  mpkvList.size()>j*100+100?j*100+100: mpkvList.size() );
				
				Object vs=slist.stream().reduce("",(t1,t2) -> t1+",'"+t2+"'");
				
				minorSlq.append(vs.toString().replaceFirst(",", "")).append(")");
				
				String minorSql = minorSlq.toString();
				
				List<Map<String, Object>>  minordata=QueryHelper.queryMapList(minorSource, minorSql);
				
				Map<Object, Map<String, Object >> minorDataMap =minordata.stream().collect(Collectors.toMap(it-> it.get(ipk), v->v));
				
				List<FaultsData>  list=CompareDataUtils.compareData(dataMap, minorDataMap, columnMapping);
				
				faultList.addAll(list);
		
			}
			
		}
		
		Map<String, Object> inspectResult =new HashMap<>();
		
		inspectResult.put("faultTotal", faultList.size());
		inspectResult.put("faultData", faultList);
		qr.setWarnning(faultList.size()>1?1:0); 
		qr.setInspectResult(inspectResult);
		
		
		return qr;
		
		
		 
	}
	

	
	
	
	private QualityResultPo  statistics(DataSource sSource,DataSource tSource,EtlJob job,QualityStrategyEntity stra) throws SQLException {
		
		QualityResultPo  sr =new QualityResultPo();
		sr.setStartTime(new Date());
		 
		QualityStatisticsRulePo  statisticsRule = stra.getStatisticsRule();
		
		String ssql=statisticsRule.getSourceSql();
		String tsql=statisticsRule.getTargetSql();

		String warrpedSsql=	expressionHelper.parseExpression(ssql).getValue(ExpressionHelper.map2EvalContext(stra.getParamMap()), String.class);
		
		String warrpedTsql=	expressionHelper.parseExpression(tsql).getValue(ExpressionHelper.map2EvalContext(stra.getParamMap()), String.class);
		
		List<Map<String, Object>> srcRes=	QueryHelper.queryMapList(sSource, warrpedSsql);
		
		List<Map<String, Object>> tagRes=	QueryHelper.queryMapList(tSource, warrpedTsql);
		
		 QualityHandler qualityHandler=qualityHandlerMap.get(statisticsRule.getHandler());
	
		 Map<String, Object>  inspectResult =qualityHandler.doHandler(srcRes, tagRes, stra);
		
		 sr.setStrategyId(stra.getId());
		 sr.setInspectResult(inspectResult);
		 sr.setFinishTime(new Date());			 
		 
		 Integer diffSzie= (Integer) inspectResult.get(Constant.DIFF_SIZE);
		 
		 if(diffSzie!=null&&diffSzie>0) {
			 sr.setWarnning(1);
		 }else {
			 sr.setWarnning(0);
		 }
		 
		return sr;

		
	}

	
}
